<?php 
X_config::mysql_table(X_config::$stream_dbTable, array(
    'id'        => "bigint(20)",
    'microtime' => "bigint(20)",
    'expires'   => "int(11)", 
    'from'      => "int(11)",
    'for'       => "varchar(20)",
    'action'    => "int(11)",
    'content'   => "text",
),"id","ENGINE=InnoDB CHARACTER SET utf8 COLLATE utf8_general_ci");

X_options::$stream= true;
class X_stream extends X{
    protected $database= null;
    public $defaultExpires;
    public $timeout;
    public $interval;
    public $dbTable;
    
    static function escapeJS($text,$nl2br=false){
        if($nl2br){
            $text= str_replace("\r","<br />",$text);
            $text= str_replace("\n","<br />",$text);
            $text= str_replace("\\","\\\\",$text);
            $text= str_replace("'","\\'",$text);
        }else{
            $text= str_replace("\\","\\\\",$text);
            $text= str_replace("'","\\'",$text);
            $text= str_replace("\r"," ",$text);
            $text= str_replace("\n"," ",$text);
        }
        return $text;
    }
    
    public final function __construct(X_mysql $database=null, $stream_dbTable=null, $stream_expiration=null, $stream_timeout=null, $stream_interval=null){
        $this->database= $database;
        $this->dbTable= ($stream_dbTable!==null)? $stream_dbTable : X_config::$stream_dbTable;
        $this->defaultExpires= ($stream_expiration!==null)? $stream_expiration : X_config::$stream_expiration;
        $this->timeout= ($stream_timeout!==null)? $stream_timeout : X_config::$stream_timeout;
        $this->interval= ($stream_interval!==null)? $stream_interval : X_config::$stream_interval;
    }
    
    public final function microTime(){
        return (int)((double)microtime(true)*100000000);
    }
    
    public final function getStream($for, $start=0, $action=null, $timeout=null, $interval=null){
        $this->initStream();
        $readOnlySession= $_SESSION;
        session_write_close();
        if($this->database!==null){
            $startTime= time();
            if($timeout===null)$timeout= $this->timeout;
            $endTime= $startTime+$timeout;
            if($interval===null)$interval= $this->interval;
            $this->database->delete('stream', "`expires` < ?", $startTime);
            $query= new X_mysql_query('stream',$this->database->link);
            $query->select()
                  ->where("`microtime` > ?",$start)
                  ->where("`microtime` <= ((UNIX_TIMESTAMP()+1)*100000000)")
                  ->where("`for` IN (?)", $for)
                  ->order(array('id'=>"ASC"));
            if($action!==null)$query->where("`action` IN (?)", $action);
            $query->query(); // Build statement and cache it, ready to execute multiple times
            while($endTime > time() && !connection_aborted() && ($stream= $query->fetchAll())==false) usleep($interval); // Search in database every 400ms, for max 15 secs, until a new stream is found. 
            if(is_array($stream)){
                return $this->parseStream($stream, $readOnlySession);
            }
            return null;
        }else sleep($timeout);
    }
    
    protected function initStream(){
        
    }
    
    protected function parseStream($stream){
        $streamStr= "";
        $javascript= "";
        foreach($stream as $action){
            switch($action['action']){
                default : 
                    $javascript= "alert('".X_stream::escapeJS($action['content'])."');";
                break;
            }
            $streamStr .= ";x_stream_lastMicrotime='".$action['microtime']."';".$javascript;
        }
        return $streamStr;
    }
    
    public function send($from, $for, $content, $action=0, $expires=null, $postpone=0){
        if($expires===null)$expires= $this->defaultExpires+$postpone;
        $microTime= $this->microTime()+($postpone*100000000);
        if(!is_array($for)) $for= array($for);
        $query= $this->database->newQuery($this->dbTable);
        foreach($for as $f){
            $query->insert(array(
                'microtime' => $microTime,
                'expires'   => time()+$expires,
                'from'      => $from,
                'for'       => $f,
                'action'    => $action,
                'content'   => $content,
            ));
        }
        return $query->execute();
    }
    
}
?>
<script type="text/javascript">
    var x_stream_lastMicrotime= '0';
    function getStream(params,repeat){
        if(params == undefined){
            params = new Array();
        }
        params['start']= x_stream_lastMicrotime;
        params['action']= 'stream';
        x_ajax('<?php echo X_config::$stream_url;?>',params,function(data){
            x_safeEval(data);
            if(repeat)setTimeout(function(){getStream(params,true)},<?php echo X_config::$stream_ajaxWait;?>);
        }, false);
    }
</script>
